package com.idanchuang.count.service.impl;

import com.idanchuang.component.base.exception.common.ErrorCode;
import com.idanchuang.component.base.exception.core.ExFactory;
import com.idanchuang.count.core.model.BizProperties;
import com.idanchuang.count.core.util.keyformat.UserCountKeysFormatter;
import com.idanchuang.count.service.MsgUnreadCountService;
import com.idanchuang.user.count.service.api.request.command.read.AddSysMsgCommand;
import com.idanchuang.user.count.service.api.request.command.read.AddUserMsgCommand;
import com.idanchuang.user.count.service.api.request.command.read.BatchAddUserMsgCommand;
import com.idanchuang.user.count.service.api.request.command.read.RemoveSysMsgCommand;
import com.idanchuang.user.count.service.api.request.command.read.RemoveUserMsgCommand;
import com.idanchuang.user.count.service.api.request.command.read.UserMsgAllReadCommand;
import com.idanchuang.user.count.service.api.request.query.read.BatchUserUnreadMsgCountQuery;
import com.idanchuang.user.count.service.api.request.query.read.UserMsgIfReadBatchQuery;
import com.idanchuang.user.count.service.api.request.query.read.UserUnreadMsgCountQuery;
import com.idanchuang.user.count.service.api.response.BatchUserSceneCountDTO;
import com.idanchuang.user.count.service.api.response.UserSceneCountDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import static com.idanchuang.count.core.util.CountDtoAssembler.buildBatchUserSceneCountDTO;
import static com.idanchuang.count.core.util.CountDtoAssembler.buildUserSceneCountDTO;
import static com.idanchuang.count.core.util.keyformat.UserCountKeysFormatter.formatRecentReadTimeKey;
import static com.idanchuang.count.core.util.keyformat.UserCountKeysFormatter.formatSysMsgSetKey;
import static com.idanchuang.count.core.util.keyformat.UserCountKeysFormatter.formatUserMsgSetKey;

/**
 * TODO 下面的一部分 pipeline 操作用 lua 脚本更为合适，等 gauss redis 支持 lua 之后可以改造
 *
 * @author Richard_yyf
 * @version 1.0 2021/7/7
 */
@Service
@Slf4j
public class MsgUnreadCountServiceImpl implements MsgUnreadCountService {

    private final StringRedisTemplate redisTemplate;
    private final RedisSerializer<String> serializer;

    /**
     * 用户 N 天未读，则设置未读数为0
     * 默认 90 * 86400 = 7776000
     */
    @Value("${count.msg.expireTimeInSec:7776000}")
    private Long expireTimeInSec;

    private final static int BATCH_QUERY_MSG_IF_READ_LIMIT = 20;

    @Resource
    private BizProperties bizProperties;

    public MsgUnreadCountServiceImpl(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.serializer = redisTemplate.getStringSerializer();
    }

    @Override
    public void readAll(UserMsgAllReadCommand command) {

        wrapExecutePipelined(convertRedisOperation(command));
    }

    @Override
    public void readAllInAsyncBatch(List<UserMsgAllReadCommand> readCommands) {
        wrapBatchExecutePipelined(readCommands
                .stream()
                .map(this::convertRedisOperation)
                .collect(Collectors.toList())
        );
    }

    @SuppressWarnings("all")
    private Consumer<RedisConnection> convertRedisOperation(UserMsgAllReadCommand command) {
        String userMsgSetKey = formatUserMsgSetKey(
                bizProperties.getBizId(command.getBizCode()), command.getUserCode(), command.getScene());

        // 这里不使用hash表的原因是要对单个时间设置 expire 时间
        byte[] recentReadTimeKey = serializer.serialize(
                formatRecentReadTimeKey(bizProperties.getBizId(command.getBizCode()), command.getUserCode(), command.getScene()));
        byte[] recentReadTime = serializer.serialize(String.valueOf(System.currentTimeMillis()));
        return redisConnection -> {
            // del user msg set
            redisConnection.del(serializer.serialize(userMsgSetKey));
            // update user recent read time
            redisConnection.setEx(recentReadTimeKey, expireTimeInSec, recentReadTime);
        };
    }

    @Override
    @SuppressWarnings("all")
    public void addUserMsg(AddUserMsgCommand command) {

        String userMsgSetKey = formatUserMsgSetKey(bizProperties.getBizId(command.getBizCode()), command.getUserCode(), command.getScene());
        // zset add msg, score command generate time
        if (redisTemplate.hasKey(userMsgSetKey)) {
            redisTemplate.opsForZSet().add(userMsgSetKey, command.getMsgId(), command.getTimeStamp());
        } else {
            // 如果之前 用户维度的消息列表不存在，则在初次增加未读消息之后，设置过期时间
            wrapExecutePipelined(
                    redisConnection -> {
                        byte[] key = serializer.serialize(userMsgSetKey);
                        byte[] val = serializer.serialize(command.getMsgId());
                        redisConnection.zSetCommands().zAdd(key, command.getTimeStamp(), val);
                        redisConnection.expire(key, expireTimeInSec);
                    }
            );
        }
    }

    @Override
    @SuppressWarnings("all")
    public void addUserMsgInAsyncBatch(List<AddUserMsgCommand> commands) {

        List<String> userMsgSetKeyList = commands.stream()
                .map(command -> formatUserMsgSetKey(bizProperties.getBizId(command.getBizCode()), command.getUserCode(), command.getScene()))
                .collect(Collectors.toList());
        List<String> notExistSetKeys = queryNotExistKeys(userMsgSetKeyList);

        wrapExecutePipelined(
                redisConnection -> {
                    for (int i = 0; i < userMsgSetKeyList.size(); i++) {
                        byte[] key = serializer.serialize(userMsgSetKeyList.get(i));
                        byte[] val = serializer.serialize(commands.get(i).getMsgId());
                        redisConnection.zSetCommands().zAdd(key, commands.get(i).getTimeStamp(), val);
                    }
                    // 如果之前 用户维度的消息列表不存在，则在初次增加未读消息之后，设置过期时间
                    for (String notExistSetKey : notExistSetKeys) {
                        redisConnection.expire(serializer.serialize(notExistSetKey), expireTimeInSec);
                    }
                }
        );
    }

    @SuppressWarnings("all")
    private List<String> queryNotExistKeys(List<String> userMsgSetKeyList) {
        List<String> userMsgSetKeyDistinctSet = userMsgSetKeyList.stream()
                .distinct()
                .collect(Collectors.toList());
        List<Object> results = redisTemplate.executePipelined((RedisCallback<String>) redisConnection -> {
            userMsgSetKeyDistinctSet.forEach(
                    key -> redisConnection.exists(serializer.serialize(key))
            );
            return null;
        });
        List<String> notExistSetKeys = new ArrayList<>();
        for (int i = 0; i < userMsgSetKeyDistinctSet.size(); i++) {
            if (!(Boolean) results.get(i)) {
                notExistSetKeys.add(userMsgSetKeyDistinctSet.get(i));
            }
        }
        return notExistSetKeys;
    }

    @Override
    public void batchAddUserMsg(BatchAddUserMsgCommand command) {

        Integer bizId = bizProperties.getBizId(command.getBizCode());
        List<byte[]> keys = command.getUserCodeList()
                .stream()
                .map(userCode -> serializer.serialize(formatUserMsgSetKey(bizId, userCode, command.getScene())))
                .collect(Collectors.toList());

        // val 不会为null，因为 msgId 不为空
        byte[] val = serializer.serialize(command.getMsgId());

        wrapExecutePipelined(redisConnection ->
                keys.forEach(
                        // zAdd(byte[] key, double score, byte[] value)
                        k -> redisConnection.zSetCommands().zAdd(k, command.getTimeStamp(), val)
                ));
    }

    @Override
    public void batchAddUserMsgInAsyncBatch(List<BatchAddUserMsgCommand> batchAddUserMsgCommands) {
        batchAddUserMsgCommands.forEach(this::batchAddUserMsg);
    }

    @Override
    public void addSysMsg(AddSysMsgCommand command) {
        String key = formatSysMsgSetKey(bizProperties.getBizId(command.getBizCode()), command.getScene());
        // zset add msg, score current time
        redisTemplate.opsForZSet().add(key, command.getMsgId(), command.getTimeStamp());
    }

    @Override
    public void removeSysMsg(RemoveSysMsgCommand command) {
        // zset rm msg
        String key = formatSysMsgSetKey(bizProperties.getBizId(command.getBizCode()), command.getScene());
        redisTemplate.opsForZSet().remove(key, command.getMsgId());
    }

    @Override
    public void removeUserMsg(RemoveUserMsgCommand command) {
        // zset rm msg
        String key = formatUserMsgSetKey(bizProperties.getBizId(command.getBizCode()), command.getUserCode(), command.getScene());
        redisTemplate.opsForZSet().remove(key, command.getMsgIdList().toArray());
    }

    @Override
    public BatchUserSceneCountDTO batchQueryUnreadMsgCount(BatchUserUnreadMsgCountQuery query) {

        Integer bizId = bizProperties.getBizId(query.getBizCode());
        Integer userCode = query.getUserCode();
        List<String> scenes = query.getScenes().stream().distinct().collect(Collectors.toList());

        //  won't be null here, zset zcount will return 0 if key not exist
        List<Integer> userUnreadList = queryUserUnreadCountList(bizId, userCode, scenes);
        List<Integer> sysUnreadList = querySysUnreadCountList(bizId, userCode, scenes);

        return buildBatchUserSceneCountDTO(userCode, scenes, userUnreadList, sysUnreadList);
    }

    @Override
    public List<String> batchQueryUserMsgIfRead(UserMsgIfReadBatchQuery query) {

        if (CollectionUtils.isEmpty(query.getMsgNos())) {
            return Collections.emptyList();
        }

        if (query.getMsgNos().size() > BATCH_QUERY_MSG_IF_READ_LIMIT) {
            throw ExFactory.throwWith(ErrorCode.PARAM_ERROR, "msgNos size over limit, <=" + BATCH_QUERY_MSG_IF_READ_LIMIT);
        }

        Integer bizId = bizProperties.getBizId(query.getBizCode());
        List<String> msgNos = query.getMsgNos();
        Integer userCode = query.getUserCode();
        String scene = query.getScene();

        List<String> unreadMsgNos = new ArrayList<>(msgNos.size());

        Set<String> unreadUserMsgSet = queryUnreadUserMsgSet(bizId, userCode, scene);
        filterByMsgSet(unreadMsgNos, msgNos, unreadUserMsgSet);
        if (msgNos.isEmpty()) {
            return unreadMsgNos;
        }

        Set<String> unreadSysMsgSet = queryUnreadSysMsgSet(bizId, userCode, scene);
        filterByMsgSet(unreadMsgNos, msgNos, unreadSysMsgSet);
        return unreadMsgNos;
    }

    private Set<String> queryUnreadSysMsgSet(Integer bizId, Integer userCode, String scene) {
        long now = System.currentTimeMillis();
        String sysMsgSetKey = formatSysMsgSetKey(bizId, scene);
        String recentReadTimeKey = formatRecentReadTimeKey(bizId, userCode, scene);
        String recentReadTimeValueStr = redisTemplate.opsForValue().get(recentReadTimeKey);
        long recentReadTime = recentReadTimeValueStr == null ?
                (now - (expireTimeInSec * 1000)) : Long.parseLong(recentReadTimeValueStr);
        Set<String> unreadSysMsgSet = redisTemplate.opsForZSet().rangeByScore(sysMsgSetKey, recentReadTime, now);
        if (!CollectionUtils.isEmpty(unreadSysMsgSet)) {
            log.info("queryUnreadSysMsgSet, size: {}", unreadSysMsgSet.size());
        }
        return unreadSysMsgSet;
    }

    private void filterByMsgSet(List<String> unreadMsgNos, List<String> sourceMsgNos, Set<String> msgNoSetForFilter) {
        if (CollectionUtils.isEmpty(msgNoSetForFilter)) {
            return;
        }
        Iterator<String> msgNoIterator = sourceMsgNos.iterator();
        int count = 0;
        while (msgNoIterator.hasNext()) {
            String msgNo = msgNoIterator.next();
            if (msgNoSetForFilter.contains(msgNo)) {
                unreadMsgNos.add(msgNo);
                msgNoIterator.remove();
                count++;
            }
            if (count >= msgNoSetForFilter.size()) {
                // 说明已经把 unreadUserMsgSet 中的未读消息都加进来了，没必要再查unreadUserMsgSet了
                break;
            }
        }
    }

    private Set<String> queryUnreadUserMsgSet(Integer bizId, Integer userCode, String scene) {
        Set<String> unreadUserMsgSet;
        String userMsgSetKey = formatUserMsgSetKey(bizId, userCode, scene);
        Long userMsgUnreadCount = redisTemplate.opsForZSet().zCard(userMsgSetKey);
        if (userMsgUnreadCount != null && userMsgUnreadCount > 0) {
            // get all
            unreadUserMsgSet = redisTemplate.opsForZSet().range(userMsgSetKey, 0, -1);
        } else {
            unreadUserMsgSet = Collections.emptySet();
        }
        if (!CollectionUtils.isEmpty(unreadUserMsgSet)) {
            log.info("queryUnreadUserMsgSet, size: {}", unreadUserMsgSet.size());
        }
        return unreadUserMsgSet;
    }

    @SuppressWarnings("all")
    private List<Integer> querySysUnreadCountList(Integer bizId, Integer userCode, List<String> scenes) {

        List<String> sysMsgSetKeyList = new ArrayList<>(scenes.size());
        List<String> recentReadTimekeys = new ArrayList<>(scenes.size());
        for (int i = 0; i < scenes.size(); i++) {
            sysMsgSetKeyList.add(formatSysMsgSetKey(bizId, scenes.get(i)));
            recentReadTimekeys.add(formatRecentReadTimeKey(bizId, userCode, scenes.get(i)));
        }

        List<Object> recentReadTimeStrList = redisTemplate.executePipelined((RedisCallback<String>) redisConnection -> {
            recentReadTimekeys.forEach(
                    recentReadTimekey -> redisConnection.get(serializer.serialize(recentReadTimekey))
            );
            return null;
        });
        long now = System.currentTimeMillis();
        List<Long> recentReadTimeList = recentReadTimeStrList.stream()
                .map(recentReadTimeStr -> (recentReadTimeStr == null ?
                        (now - (expireTimeInSec * 1000)) : Long.parseLong((String) recentReadTimeStr)))
                .collect(Collectors.toList());

        List<Object> results = redisTemplate.executePipelined((RedisCallback<String>) redisConnection -> {
            for (int i = 0; i < sysMsgSetKeyList.size(); i++) {
                redisConnection.zSetCommands().zCount(serializer.serialize(sysMsgSetKeyList.get(i)), recentReadTimeList.get(i), now);
            }
            return null;
        });

        return results.stream()
                .map(count -> Math.toIntExact((Long) count))
                .collect(Collectors.toList());
    }

    @SuppressWarnings("all")
    private List<Integer> queryUserUnreadCountList(Integer bizId, Integer userCode, List<String> scenes) {
        List<String> keys = scenes.stream()
                .map(scene -> UserCountKeysFormatter.formatUserMsgSetKey(bizId, userCode, scene))
                .collect(Collectors.toList());;

        List<Object> results = redisTemplate.executePipelined((RedisCallback<String>) redisConnection -> {
            keys.forEach(
                    key -> redisConnection.zSetCommands().zCard(serializer.serialize(key))
            );
            return null;
        });

        return results.stream()
                .map(count -> Math.toIntExact((Long) count))
                .collect(Collectors.toList());
    }

    @Override
    public UserSceneCountDTO queryUnreadMsgCount(UserUnreadMsgCountQuery query) {

        Integer bizId = bizProperties.getBizId(query.getBizCode());

        int userUnread = queryUserUnreadCount(bizId, query.getUserCode(), query.getScene());

        int sysUnread = querySysUnreadCount(bizId, query.getUserCode(), query.getScene());

        return buildUserSceneCountDTO(query.getUserCode(), query.getScene(), sysUnread + userUnread);
    }

    @SuppressWarnings("all")
    private int queryUserUnreadCount(Integer bizId, Integer userCode, String scene) {

        String userMsgSetKey = formatUserMsgSetKey(bizId, userCode, scene);
        // user unread = zset zcard
        //  won't be null here, will return 0 if key not exist
        return Math.toIntExact(redisTemplate.opsForZSet().zCard(userMsgSetKey));
    }

    @SuppressWarnings("all")
    private int querySysUnreadCount(Integer bizId, Integer userCode, String scene) {
        String recentReadTimeKey = formatRecentReadTimeKey(bizId, userCode, scene);
        String recentReadTimeInStr = redisTemplate.opsForValue().get(recentReadTimeKey);
        String sysMsgSetKey = formatSysMsgSetKey(bizId, scene);

        // 当最近一次已读的时间不在时，直接按照查最近一段时间的系统消息数
        long now = System.currentTimeMillis();
        long recentReadTime = recentReadTimeInStr == null ?
                (now - (expireTimeInSec * 1000)) : Long.parseLong(recentReadTimeInStr);
        // sys unread = zset zcount
        // won't be null here
        return Math.toIntExact(redisTemplate.opsForZSet().count(sysMsgSetKey, recentReadTime, System.currentTimeMillis()));
    }

    private void wrapBatchExecutePipelined(List<Consumer<RedisConnection>> operations) {
        redisTemplate.executePipelined((RedisCallback<String>) redisConnection -> {
            operations.forEach(operation -> operation.accept(redisConnection));
            return null;
        });
    }

    private void wrapExecutePipelined(Consumer<RedisConnection> operation) {
        redisTemplate.executePipelined((RedisCallback<String>) redisConnection -> {
            operation.accept(redisConnection);
            return null;
        });
    }
}
